package com.huatai.datacenter.controller.topicmonitor.kafka;

import com.huatai.common.api.R;
import com.huatai.datacenter.entity.messagequeuemonitor.LagRecordQueryDTO;
import com.huatai.datacenter.entity.messagequeuemonitor.LagStatVO;
import com.huatai.datacenter.entity.messagequeuemonitor.PageResult;
import com.huatai.datacenter.service.KafkaMonitorService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;

/**
 * 用于消息队列消费情况监控的Controller层，Kafka的
 * 层级关系：主题-消费组-分区-堆积消息
 *
 * @author lizhong
 * @date 2023/7/10
 */
@Slf4j
@RestController("KafkaTopicMonitorController")
@RequestMapping("/MQMonitor/kafka") /*MQ = MessageQueue消息队列*/
@Api(value = "Kafka消息队列消费监控", tags = "Kafka消息队列消费监控")
public class TopicMonitorController {

	@Autowired
	private KafkaMonitorService kafkaMonitorService;

	/**
	 * 获取Kafka所有主题（不包括系统的主题）
	 *
	 * @return 所有主题的列表Set
	 * @author lizhong
	 */
	/*@GetMapping("/allTopicsKafka")
	@ApiOperation("(不用)获取Kafka所有主题")
	public R<Set<String>> allTopicKafka(@RequestParam String clusterID) {
		try {
			return R.data(kafkaMonitorService.getAllTopic(clusterID));
		} catch (Exception e) {
			log.error("get all topics error, ", e);
		}
		return R.fail("获取集群" + clusterID + "所有主题错误! Please check");
	}*/

	/**
	 * 获取一个集群下的所有堆积情况 Kafka
	 * 原理：直接读取缓存，为空也直接读
	 * FIXME: 要根据实际生成时间调整定时任务间隔时间！
	 *
	 * @param clusterID 集群id
	 * @return
	 * @author lizhong
	 * @date 2023年8月2日
	 */
	@ApiOperation("获取一个cluster下所有主题消费情况的列表，首页展示")
	@GetMapping
	public R<PageResult> list(@ApiParam(required = true) @RequestParam String clusterID) {
		try {
			log.info("首页展示所有主题消费情况");
			List<LagStatVO> lagStatVOListCache = kafkaMonitorService.getLagStatVOListCache(clusterID);
			log.info((lagStatVOListCache.size() == 0) ?
				"读到的所有主题消费情况缓存为空，定时任务未完成" :
				"读到的所有主题消费情况缓存正常，定时任务完成");
			return R.data(new PageResult(lagStatVOListCache.size(), lagStatVOListCache));
		} catch (Exception e) {
			log.error("获取所有主题消费情况的列表错误, ", e);
		}
		return R.fail("获取集群" + clusterID + "所有主题消费情况的列表错误! 可能是远程kafka服务器的问题");
	}

	/**
	 * 获取所有堆积消息 Kafka Test
	 *
	 * @param clusterID 集群id
	 * @return
	 * @author lizhong
	 */
	@ApiOperation("(不用)获取所有主题消费情况的列表，首页展示(不走缓存)")
	@GetMapping("/test")
	public R listTest(@ApiParam(required = true) @RequestParam String clusterID) {
		try {
			log.info("首页展示所有主题消费情况");
			return R.data(kafkaMonitorService.listAllConsumeStatKafka(clusterID));
		} catch (Exception e) {
			log.error("获取所有主题消费情况的列表错误, ", e);
		}
		return R.fail("获取集群" + clusterID + "所有主题消费情况的列表错误! 可能是远程kafka服务器的问题");
	}

	/**
	 * 获取一个(topic, partition, group)下的所有堆积消息 Kafka
	 *
	 * @param lagRecordQueryDTO (topic, partition, group)
	 * @return 所有堆积消息
	 * @author lizhong
	 */
	@ApiOperation("获取详细堆积消息 Kafka")
	@GetMapping("/getLagRecordKafka")
	public R getLagRecordListKafka(LagRecordQueryDTO lagRecordQueryDTO) {
		try {
			log.info("堆积消息查询: {}", lagRecordQueryDTO);
			return R.data(kafkaMonitorService.getLagRecordListKafka(lagRecordQueryDTO));
		} catch (Exception e) {
			log.error("获取堆积消息错误, ", e);
		}
		return R.fail("获取堆积消息错误! Please check");
	}

	/**
	 * 测试用
	 *
	 * @param clusterID
	 * @return
	 */
	/*@ApiOperation("(不用)getAllTopicPartitionState Kafka")
	@GetMapping("/getAllTopicPartitionState")
	public R<PageResult> getAllTopicPartitionState(@RequestParam String clusterID) {
		try {
			return R.data(kafkaMonitorService.getAllTopicPartitionState(clusterID));
		} catch (Exception e) {
			log.error("获取主题列表错误, ", e);
		}
		return R.fail("获取主题列表错误! Please check");
	}*/
}
